package com.kafka.service;

import com.kafka.rest.CreateProductRestModel;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.example.ProductCreatedEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;

import java.nio.charset.StandardCharsets;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

/**
 * @author Administrator
 */

@Service
public class ProductServiceImpl implements ProductService {
    private static final Logger LOGGER = LoggerFactory.getLogger(ProductServiceImpl.class);
    private KafkaTemplate<String, ProductCreatedEvent> kafkaTemplate;

    public ProductServiceImpl(KafkaTemplate<String, ProductCreatedEvent> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    @Override
    public String createProduct(CreateProductRestModel model) throws ExecutionException, InterruptedException {
        String productId = UUID.randomUUID().toString();
        ProductCreatedEvent event = new ProductCreatedEvent(
                productId, model.getTitle(), model.getPrice(), model.getQuantity()
        );
        ProducerRecord<String, ProductCreatedEvent> record = new ProducerRecord<>(
                "huran-test", productId, event
        );
        record.headers().add("messageId", UUID.randomUUID().toString().getBytes(StandardCharsets.UTF_8));
        //等待kafka
        kafkaTemplate.send(record).get();

//        send.whenComplete((result, ex) -> {
//            if (ex != null) {
//                LOGGER.error("发送消息失败:" + ex.getMessage());
//            } else {
//                LOGGER.info("发送消息成功:" + result.getRecordMetadata());
//            }
//        });
        //  SendResult<String, ProductCreatedEvent> join = send.join();
        LOGGER.info("返回产品ID:" + productId);

        return productId;
    }

    @Override
    public String createProductSync(CreateProductRestModel model) throws ExecutionException, InterruptedException {
        String productId = UUID.randomUUID().toString();
        ProductCreatedEvent event = new ProductCreatedEvent(
                productId, model.getTitle(), model.getPrice(), model.getQuantity()
        );
        //等待kafka


        SendResult<String, ProductCreatedEvent> stringProductCreatedEventSendResult = kafkaTemplate.send("huran-test", productId, event).get();

        //  RecordMetadata recordMetadata = stringProductCreatedEventSendResult.getRecordMetadata();
        // LOGGER.info("============>>>partition:{}",recordMetadata.partition());
        // LOGGER.info("============>>>topic:{}",recordMetadata.topic());
        // LOGGER.info("============>>>timestamp:{}",recordMetadata.timestamp());
        // LOGGER.info("============>>>offset:{}",recordMetadata.offset());


        //LOGGER.info("返回产品ID:"+productId);
        return productId;
    }
}
